package com.nucarf.bi.digtalize.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

@Slf4j
public class CdcJsonDeserializationSchema implements DebeziumDeserializationSchema<JSONObject> {
    @Override
    public void deserialize(SourceRecord record, Collector<JSONObject> collector) throws Exception {
        try {
            Struct dataRecord = (Struct) record.value();

            Struct afterStruct = dataRecord.getStruct("after");
            Struct beforeStruct = dataRecord.getStruct("before");
        /*
          todo 1，同时存在 beforeStruct 跟 afterStruct数据的话，就代表是update的数据
               2,只存在 beforeStruct 就是delete数据
               3，只存在 afterStruct数据 就是insert数据
         */

            JSONObject logJson = new JSONObject();
            JSONObject beforeJson = new JSONObject();
            JSONObject afterJson = new JSONObject();
            JSONObject prikeyJson = new JSONObject();

            String canal_type = "";
            List<Field> fieldsList = null;
            if (afterStruct != null && beforeStruct != null) { //修改数据
                canal_type = "update";
            } else if (afterStruct != null) {   //新增数据
                canal_type = "insert";
            } else if (beforeStruct != null) {   //删除数据
                canal_type = "detele";
            } else {
                System.out.println("未知操作={" + JSON.toJSONString(record) + "}");
            }

            if (beforeStruct != null) {
                fieldsList = beforeStruct.schema().fields();
                for (Field field : fieldsList) {
                    String fieldName = field.name();
                    Object fieldValue = beforeStruct.get(fieldName);
                    beforeJson.put(fieldName, fieldValue);
                }
            }
            if (afterStruct != null) {
                fieldsList = afterStruct.schema().fields();
                //todo 字段与值
                for (Field field : fieldsList) {
                    String fieldName = field.name();
                    Object fieldValue = afterStruct.get(fieldName);
                    afterJson.put(fieldName, fieldValue);
                }
            }
            logJson.put("before", beforeJson);
            logJson.put("after", afterJson);
            //todo 拿到databases table信息
            Struct source = dataRecord.getStruct("source");
            Object db = source.get("db");
            Object table = source.get("table");
            Object ts_ms = source.get("ts_ms");

            logJson.put("database", db);
            logJson.put("table", table);
            logJson.put("ts", ts_ms);
            logJson.put("op_type", canal_type);

            //topic
            log.debug("topic = " + record.topic());

            //主键字段
            Struct pk = (Struct) record.key();
            if(pk != null){
                List<Field> pkFieldList = pk.schema().fields();
                int partitionerNum = 0;
                for (Field field : pkFieldList) {
                    prikeyJson.put(field.name(), pk.get(field.name()));
                    Object pkValue = pk.get(field.name());
                    partitionerNum += pkValue.hashCode();
                }
                int hash = Math.abs(partitionerNum) % 3;
                logJson.put("pk_hashcode", hash);
                logJson.put("primkey", prikeyJson);
            }
            collector.collect(logJson);
        }catch (Exception e){
            e.printStackTrace();
        }
    }


    @Override
    public TypeInformation<JSONObject> getProducedType() {
        return BasicTypeInfo.of(JSONObject.class);
    }
}
